package com.briup.prs;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.KeyValueLineRecordReader;
import org.apache.hadoop.mapreduce.lib.input.KeyValueTextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import java.io.IOException;
import java.util.ArrayList;

/**
 * @author adam
 * @date 2022/2/8
 * 目的：得到商品的共现关系=》商品的相似
 */
public class Step3 extends Configured implements Tool {

    public static class S3Mapper extends Mapper<Text, Text, Text, Text> {
        @Override
//        key=userid:shopid    value=偏好
        protected void map(Text key, Text value, Context context) throws IOException, InterruptedException {
//            data=[userid,shopid]
            String[] data = key.toString().split(":");
            //将用户id作为输出的key
            Text outkey = new Text(data[0]);
            //将商品id作为输出的value
            Text outvalue = new Text(data[1]);
//            key:userId   value:shopId
            context.write(outkey, outvalue);
        }
    }

    public static class S3Reducer extends Reducer<Text, Text, Text, Text> {
        @Override
//        key：useId     values:[shopId,shopId,shopId,shopId,shopId]
        protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {
//            准备接收用户偏好的所有商品
            ArrayList<String> list = new ArrayList<>();
            for (Text value : values) {
                list.add(value.toString());
            }
            for (String s : list) {
                for (String s1 : list) {
                    if (s.equals(s1)){
                        continue;
                    }
//                    key:shopid  value:shopid
                    context.write(new Text(s),new Text(s1));
                }
            }
        }
    }

    @Override
    public int run(String[] strings) throws Exception {
        Configuration conf = getConf();
        String in = conf.get("in");
        String out = conf.get("out");
        //设置分隔符
        conf.set(KeyValueLineRecordReader.KEY_VALUE_SEPARATOR, "\t");
        Job job = Job.getInstance(conf, "step3");
        job.setJarByClass(Step3.class);

        job.setInputFormatClass(KeyValueTextInputFormat.class);
        KeyValueTextInputFormat.addInputPath(job,
                new Path(in));

        job.setMapperClass(S3Mapper.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(Text.class);

        job.setReducerClass(S3Reducer.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        job.setOutputFormatClass(TextOutputFormat.class);
        TextOutputFormat.setOutputPath(job,
                new Path(out));
        job.waitForCompletion(true);
        return 0;
    }

    public static void main(String[] args) throws Exception {
        ToolRunner.run(new Step3(),args);
    }
}
